package hamster.csustef.acquisition.flink.sink.serialization;

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;

/**
 * kafka解析
 *
 * @author yin hamaster
 * @date 2022-06-14
 */
public class StringSerializationSchema
        implements KafkaSerializationSchema<String> {
    private String topic;

    public StringSerializationSchema(String topic) {
        this.topic = topic;
    }

    @Override
    public void open(SerializationSchema.InitializationContext context) throws Exception {

    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long aLong) {
        return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
    }


}